package org.snail.transport.handler;

import com.google.common.base.Throwables;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.Signal;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.snail.transport.ConnectorDispatcher;
import org.snail.transport.impl.NettyConnector;
import org.snail.transport.protocol.Message;

import java.io.IOException;


/**
 * 客户端响应逻辑处理
 *
 * @author fish
 */
@ChannelHandler.Sharable
public class ConnectorHandler extends ChannelInboundHandlerAdapter {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(ConnectorHandler.class);

    private NettyConnector connector;

    private ConnectorDispatcher dispatcher;

    public ConnectorHandler(NettyConnector connector){
        this.connector = connector;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        dispatcher.dispatch((Message)msg);
        ReferenceCountUtil.release(msg);

    }


    /**
     * 高水位线: ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK
     * 低水位线: ChannelOption.WRITE_BUFFER_LOW_WATER_MARK
     */
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        ChannelConfig config = channel.config();
        if (channel.isWritable()) {
            // 曾经高于高水位线的OutboundBuffer现在已经低于WRITE_BUFFER_LOW_WATER_MARK了
            if (logger.isWarnEnabled()) {
                logger.warn("{} is writable(rehabilitate), low water mask: {}, the number of flushed entries that are not written yet: {}.",
                        channel,
                        config.getWriteBufferLowWaterMark(), channel.unsafe().outboundBuffer().size());
            }
            config.setAutoRead(true);
        } else {
            // 当前channel的缓冲区(OutboundBuffer)大小超过了WRITE_BUFFER_HIGH_WATER_MARK
            if (logger.isWarnEnabled()) {
                logger.warn("{} is not writable, high water mask: {}, the number of flushed entries that are not written yet: {}.",
                        channel,
                        config.getWriteBufferHighWaterMark(), channel.unsafe().outboundBuffer().size());
            }
            config.setAutoRead(false);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel channel = ctx.channel();
        if (cause instanceof Signal) {
            logger.error("An I/O signal was caught: {}, force to close channel: {}.", ((Signal) cause).name(), channel);
            channel.close();
        } else if (cause instanceof IOException) {
            logger.error("An I/O exception was caught: {}, force to close channel: {}.", Throwables.getStackTraceAsString(cause), channel);
            channel.close();
        } else {
            logger.error("An unexpected exception was caught: {}, channel: {}.", Throwables.getStackTraceAsString(cause), channel);
        }
    }


    public void setDispatcher(ConnectorDispatcher dispatcher){
        this.dispatcher = dispatcher;
    }

}
